package net.bwie.dwd;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;

// 退后行为汇总

public class Dwd_retreat_behavior {
    public static void main(String[] args) {

        // 1.创建环境
        TableEnvironment tableEnv = getTableEnv();


        // 2.创建输入表
        createInputTable(tableEnv);

        // 3.创建输出表
        createoutputTable(tableEnv);

        // 4.处理数据
        Table resultTable = handle(tableEnv);

//        tableEnv.createTemporaryView("dwd_retreat_behavior", resultTable);
//        tableEnv.executeSql("SELECT * FROM dwd_retreat_behavior").print();

        // 5.保存数据
        saveToSink(tableEnv, resultTable);

    }

    private static TableEnvironment getTableEnv() {
        EnvironmentSettings build = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tableEnvironment = TableEnvironment.create(build);

        Configuration configuration = tableEnvironment.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        configuration.setString("execution.checkpointing.interval", "5 s");

        return tableEnvironment;
    }

    private static void createInputTable(TableEnvironment tableEnv) {


        //user（消费者信息表）
        tableEnv.executeSql("CREATE TABLE ods_user (\n" +
                "user_id STRING,\n" +
                "user_name STRING,\n" +
                "user_age INT,\n" +
                "user_sex STRING\n" +
                ") WITH (\n" +
                "'connector' = 'doris',\n" +
                "'fenodes' = 'node102:8030',\n" +
                "'table.identifier' = 'transactions_ods.user',\n" +
                "'username' = 'root',\n" +
                "'password' = '123456'\n" +
                ")");
//        TableResult tableResult1 = tableEnv.executeSql("SELECT * FROM ods_user");
//        tableResult1.print();

        //order_base（订单基础信息表）
        tableEnv.executeSql("CREATE TABLE  ods_order_base (\n" +
        "order_id STRING, \n"+
        "refund_id STRING, \n"+
        "user_id STRING, \n"+
        "shop_id STRING, \n"+
        "product_id STRING, \n"+
        "pay_time TIMESTAMP(3), \n"+
        "promise_ship_time TIMESTAMP(3), \n"+
        "click_ship_time TIMESTAMP(3), \n"+
        "order_amount DOUBLE, \n"+
        "order_status STRING\n"+
        ")WITH(\n"+
        "'connector' = 'kafka',\n" +
        "'topic' = 'ods-order-base-Log',\n" +
        "'properties.bootstrap.servers' = 'node101:9092',\n" +
        "'properties.group.id' = 'ods-order-base-Log-id10',\n" +
        "'scan.startup.mode' = 'earliest-offset',\n" +
        "'format' = 'json'\n" +
        ")"
        );
//        TableResult tableResult2 = tableEnv.executeSql("SELECT * FROM ods_order_base");
//        tableResult2.print();

        //refund_detail（退款明细信息表）
        tableEnv.executeSql("CREATE TABLE ods_refund_detail (\n" +
        "refund_detail_id STRING, \n"+
        "refund_id STRING, \n"+
        "order_id STRING, \n"+
        "user_id STRING, \n"+
        "apply_time TIMESTAMP(3), \n"+
        "refund_status STRING, \n"+
        "finish_time TIMESTAMP(3), \n"+
        "product_id STRING, \n"+
        "refund_amount DOUBLE, \n"+
        "user_fill_reason STRING, \n"+
        "refund_scene STRING \n"+
        ")WITH(\n"+
        "'connector' = 'kafka',\n" +
        "'topic' = 'ods-refund-detail-log',\n" +
        "'properties.bootstrap.servers' = 'node101:9092',\n" +
        "'properties.group.id' = 'ods-refund-detail-log-id10',\n" +
        "'scan.startup.mode' = 'earliest-offset',\n" +
        "'format' = 'json'\n" +
        ")"
        );
//        TableResult tableResult3 = tableEnv.executeSql("SELECT * FROM ods_refund_detail");
//        tableResult3.print();

        //consumer_behavior（消费者行为表）
        tableEnv.executeSql("CREATE TABLE ods_consumer_behavior (\n" +
        "behavior_id STRING, \n"+
        "user_id STRING, \n"+
        "behavior_type STRING, \n"+
        "behavior_time TIMESTAMP(3), \n"+
        "behavior_scenario STRING \n"+
        ")WITH(\n"+
        "'connector' = 'kafka',\n" +
        "'topic' = 'ods-consumer-behavior-log',\n" +
        "'properties.bootstrap.servers' = 'node101:9092',\n" +
        "'properties.group.id' = 'ods-consumer-behavior-log-id10',\n" +
        "'scan.startup.mode' = 'earliest-offset',\n" +
        "'format' = 'json'\n" +
        ")"
        );
//        TableResult tableResult4 = tableEnv.executeSql("SELECT * FROM ods_consumer_behavior");
//        tableResult4.print();

    }

    private static void createoutputTable(TableEnvironment tableEnv) {

        //dwd_retreat_behavior（退后行为汇总表）
            tableEnv.executeSql("CREATE TABLE dwd_retreat_behavior (\n" +
                    //订单id
                    "order_id STRING ,\n" +
                     //  商品ID
                    "product_id STRING ,\n" +
                     // 订单支付时间
                    "pay_time TIMESTAMP(3) ,\n" +
                    //（订单总金额）
                    "order_amount DOUBLE,\n"+

                    // 退款id
                    "refund_id STRING, \n"+
                     // 退款申请时间
                    "apply_time TIMESTAMP(3) ,\n" +
                     //  退款场景:“退货退款”“已收货仅退款”“未收货退款”“未发货仅退款”
                    "refund_scene STRING ,\n" +
                     // 实际退款金额
                    "refund_amount DOUBLE ,\n" +
                    // (退款状态) /申请中/处理中/申请成功/申请关闭
                    "refund_status STRING,\n"+

                     //消费者行为id
                    "behavior_id STRING ,\n" +
                     //消费者id
                    "user_id STRING ,\n" +
                    //行为类型
                    "behavior_type STRING, \n" +
                    "PRIMARY KEY (order_id) NOT ENFORCED\n"+
                    ")WITH(\n" +
                    "'connector' = 'upsert-kafka',\n" +
                    "'topic' = 'dwd_retreat_behavior',\n" +
                    "'properties.bootstrap.servers' = 'node101:9092',\n" +
                    "'key.format' = 'json',\n" +
                    "'value.format' = 'json'\n" +
                    ")");
    }

    private static Table handle(TableEnvironment tableEnv) {

        Table table = tableEnv.sqlQuery("SELECT \n" +
                "ods_order_base.order_id,\n" +
                "ods_order_base.product_id,\n" +
                "ods_order_base.pay_time,\n" +
                "ods_order_base.order_amount,\n" +

                "ods_refund_detail.refund_id,\n" +
                "ods_refund_detail.apply_time,\n" +
                "ods_refund_detail.refund_scene,\n" +
                "ods_refund_detail.refund_amount,\n" +
                "ods_refund_detail.refund_status,\n" +

                "ods_consumer_behavior.behavior_id,\n" +
                "ods_consumer_behavior.user_id,\n" +
                "ods_consumer_behavior.behavior_type\n" +

                //order_base（订单基础信息表）
                "FROM ods_order_base " +
                 //refund_detail（退款明细信息表）
                "left join ods_refund_detail on ods_order_base.refund_id = ods_refund_detail.refund_id \n"+
                 //consumer_behavior（消费者行为表）
                "left join ods_consumer_behavior on ods_consumer_behavior.user_id = ods_order_base.user_id \n"+
                 //user（消费者信息表）
                "left join ods_user \n" +
//                "FOR SYSTEM_TIME AS OF ods_order_base.pay_time t\n" +
                " on ods_order_base.user_id = ods_user.user_id \n" +
                "where ods_refund_detail.apply_time  >=  ods_order_base.pay_time \n"
        );
        return table;
    }

    private static void saveToSink(TableEnvironment tableEnv, Table resultTable) {
        tableEnv.createTemporaryView("resultTable", resultTable);
        tableEnv.executeSql("INSERT INTO dwd_retreat_behavior SELECT * FROM resultTable");
    }

}
